#include "sys.h"
#include "net.h"
#include "log.h"
#include "thread_pool.h"
#include "cJSON.h"
#include "dial_strategy.h"
#include "devinfo.h"
#include "comm_param.h"
#include "data_report.h"
#include "heart.h"

static uint32_t currentMDay = 0, sequenceId = 0;

void dial_task_set_flag(Dial_Task_t* task, int flag)
{
    task->flag = flag;
}
void dial_task_all_set_flag(struct Dial_Task* taskall, int flag)
{
    struct Dial_Task_Head* head = &taskall->head;
    struct Dial_Task_s* ptr;
    TAILQ_FOREACH(ptr, head, entry)
    {
        dial_task_set_flag(ptr, flag);
    }
}

static void _parse_ping_tactics(cJSON* json, struct pingTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    pingTactics_t* ping_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        ping_tactics = calloc(1, sizeof(pingTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "host");
        strcpy(ping_tactics->host, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(ping_tactics->type, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        ping_tactics->port = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "size");
        ping_tactics->size = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "times");
        ping_tactics->times = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        ping_tactics->timeout = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(ping_tactics->businessTag, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(ping_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, ping_tactics, entry);
    }
}
static void _parse_tcp_tactics(cJSON* json, struct tcpTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    tcpTactics_t* tcp_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        tcp_tactics = calloc(1, sizeof(tcpTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "tcpUrlVal");
        strcpy(tcp_tactics->host, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        tcp_tactics->port = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        tcp_tactics->timeout = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(tcp_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(tcp_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, tcp_tactics, entry);
    }
}
static void _parse_dns_tactics(cJSON* json, struct dnsTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    dnsTactics_t* dns_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        dns_tactics = calloc(1, sizeof(dnsTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "ip");
        strcpy(dns_tactics->ip, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        dns_tactics->port = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        dns_tactics->timeout = json_ptr->valueint;
        json_ptr = cJSON_GetObjectItem(json_obj, "domainName");
        strcpy(dns_tactics->domainName, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(dns_tactics->type, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(dns_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(dns_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, dns_tactics, entry);
    }
}
static void _parse_http_tactics(cJSON* json, struct httpTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    httpTactics_t* http_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        http_tactics = calloc(1, sizeof(httpTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "httpUrlVal");
        strcpy(http_tactics->url, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "port");
        http_tactics->port = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        http_tactics->timeout = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(http_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(http_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, http_tactics, entry);
    }
}
static void _parse_traceroute_tactics(cJSON* json, struct tracerouteTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    tracerouteTactics_t* traceroute_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        traceroute_tactics = calloc(1, sizeof(tracerouteTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);
        json_ptr = cJSON_GetObjectItem(json_obj, "toAddress");
        strcpy(traceroute_tactics->toAddress, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "type");
        strcpy(traceroute_tactics->type, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "size");
        traceroute_tactics->size = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "maxNum");
        traceroute_tactics->maxNum = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        traceroute_tactics->timeout = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(traceroute_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(traceroute_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, traceroute_tactics, entry);
    }
}
static void _parse_httpvideodown_tactics(cJSON* json, struct httpvideodownTactics_head* head)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    httpvideodownTactics_t* httpvideodown_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        httpvideodown_tactics = calloc(1, sizeof(httpvideodownTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);

        json_ptr = cJSON_GetObjectItem(json_obj, "urls");
        if (json_ptr)
            strcpy(httpvideodown_tactics->urls, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "threads");
        httpvideodown_tactics->threads = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "downTime");
        httpvideodown_tactics->downTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "ignoreTime");
        httpvideodown_tactics->ignoreTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "threshold");
        if (json_ptr)
            httpvideodown_tactics->threshold = ((float)json_ptr->valueint) / 100.0;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        if (json_ptr)
            strcpy(httpvideodown_tactics->businessTag, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(httpvideodown_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, httpvideodown_tactics, entry);
    }
}
static void _parse_httpspeed_tactics(cJSON* json, struct httpspeedTactics_head* head, char* type)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    httpspeedTactics_t* httpspeed_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        httpspeed_tactics = calloc(1, sizeof(httpspeedTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);

        json_ptr = cJSON_GetObjectItem(json_obj, "threads");
        httpspeed_tactics->threads = json_ptr->valueint;

        if (strcmp(type, "up") == 0)
            json_ptr = cJSON_GetObjectItem(json_obj, "upTime");
        else
            json_ptr = cJSON_GetObjectItem(json_obj, "downTime");
        httpspeed_tactics->totalTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "ignoreTime");
        httpspeed_tactics->ignoreTime = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "threshold");
        if (json_ptr)
            httpspeed_tactics->threshold = ((float)json_ptr->valueint) / 100.0;

        json_ptr = cJSON_GetObjectItem(json_obj, "discardPercent");
        if (json_ptr)
            httpspeed_tactics->discardPercent = ((float)json_ptr->valueint) / 100.0;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(httpspeed_tactics->businessTag, json_ptr->valuestring);
        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(httpspeed_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, httpspeed_tactics, entry);
    }
}
static void _parse_udpspeed_tactics(cJSON* json, struct udpspeedTactics_head* head, char* type)
{
    int i = 0, arr_len = 0;
    cJSON *json_obj = NULL, *json_ptr;
    udpspeedTactics_t* udpspeed_tactics;

    arr_len = cJSON_GetArraySize(json);
    for (i = 0; i < arr_len; i++) {
        udpspeed_tactics = calloc(1, sizeof(udpspeedTactics_t));
        json_obj = cJSON_GetArrayItem(json, i);

        json_ptr = cJSON_GetObjectItem(json_obj, "ip");
        strcpy(udpspeed_tactics->ip, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "tcpPort");
        udpspeed_tactics->tcpPort = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "udpPort");
        udpspeed_tactics->udpPort = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "threads");
        udpspeed_tactics->threads = json_ptr->valueint;

        json_ptr = cJSON_GetObjectItem(json_obj, "timeout");
        udpspeed_tactics->timeout = json_ptr->valueint * 1000;

        json_ptr = cJSON_GetObjectItem(json_obj, "threshold");
        if (json_ptr)
            udpspeed_tactics->threshold = ((float)json_ptr->valueint) / 100.0;

        json_ptr = cJSON_GetObjectItem(json_obj, "discardPercent");
        if (json_ptr)
            udpspeed_tactics->discardPercent = ((float)json_ptr->valueint) / 100.0;

        json_ptr = cJSON_GetObjectItem(json_obj, "businessTag");
        strcpy(udpspeed_tactics->businessTag, json_ptr->valuestring);

        json_ptr = cJSON_GetObjectItem(json_obj, "kafkaTopic");
        if (json_ptr)
            strcpy(udpspeed_tactics->kafkaTopic, json_ptr->valuestring);
        TAILQ_INSERT_TAIL(head, udpspeed_tactics, entry);
    }
}
int dial_tactics_parse(Dial_Tactics_t* tactics, cJSON* json)
{
    cJSON* json_arr = NULL;

    TAILQ_INIT(&tactics->ping.head);
    json_arr = cJSON_GetObjectItem(json, "pingTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->ping.mutex, NULL);
        pthread_cond_init(&tactics->ping.cond, NULL);
        tactics->ping.count = cJSON_GetArraySize(json_arr);
        _parse_ping_tactics(json_arr, &tactics->ping.head);
    }

    TAILQ_INIT(&tactics->tcp.head);
    json_arr = cJSON_GetObjectItem(json, "tcpTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->tcp.mutex, NULL);
        pthread_cond_init(&tactics->tcp.cond, NULL);
        tactics->tcp.count = cJSON_GetArraySize(json_arr);
        _parse_tcp_tactics(json_arr, &tactics->tcp.head);
    }

    TAILQ_INIT(&tactics->dns.head);
    json_arr = cJSON_GetObjectItem(json, "dnsTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->dns.mutex, NULL);
        pthread_cond_init(&tactics->dns.cond, NULL);
        tactics->dns.count = cJSON_GetArraySize(json_arr);
        _parse_dns_tactics(json_arr, &tactics->dns.head);
    }

    TAILQ_INIT(&tactics->http.head);
    json_arr = cJSON_GetObjectItem(json, "httpTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->http.mutex, NULL);
        pthread_cond_init(&tactics->http.cond, NULL);
        tactics->http.count = cJSON_GetArraySize(json_arr);
        _parse_http_tactics(json_arr, &tactics->http.head);
    }

    TAILQ_INIT(&tactics->traceroute.head);
    json_arr = cJSON_GetObjectItem(json, "tracerouteTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->traceroute.mutex, NULL);
        pthread_cond_init(&tactics->traceroute.cond, NULL);
        tactics->traceroute.count = cJSON_GetArraySize(json_arr);
        _parse_traceroute_tactics(json_arr, &tactics->traceroute.head);
    }

    TAILQ_INIT(&tactics->httpvideodown.head);
    json_arr = cJSON_GetObjectItem(json, "httpvideodownTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->httpvideodown.mutex, NULL);
        pthread_cond_init(&tactics->httpvideodown.cond, NULL);
        tactics->httpvideodown.count = cJSON_GetArraySize(json_arr);
        _parse_httpvideodown_tactics(json_arr, &tactics->httpvideodown.head);
    }

    TAILQ_INIT(&tactics->httpspeed.up_head);
    json_arr = cJSON_GetObjectItem(json, "httpUploadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->httpspeed.mutex, NULL);
        pthread_cond_init(&tactics->httpspeed.cond, NULL);
        tactics->httpspeed.up_count = cJSON_GetArraySize(json_arr);
        _parse_httpspeed_tactics(json_arr, &tactics->httpspeed.up_head, "up");
    }

    TAILQ_INIT(&tactics->httpspeed.down_head);
    json_arr = cJSON_GetObjectItem(json, "httpDownloadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->httpspeed.mutex, NULL);
        pthread_cond_init(&tactics->httpspeed.cond, NULL);
        tactics->httpspeed.down_count = cJSON_GetArraySize(json_arr);
        _parse_httpspeed_tactics(json_arr, &tactics->httpspeed.down_head, "down");
    }

    TAILQ_INIT(&tactics->udpspeed.down_head);
    json_arr = cJSON_GetObjectItem(json, "udpDownloadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->udpspeed.mutex, NULL);
        pthread_cond_init(&tactics->udpspeed.cond, NULL);
        tactics->udpspeed.down_count = cJSON_GetArraySize(json_arr);
        _parse_udpspeed_tactics(json_arr, &tactics->udpspeed.down_head, "down");
    }

    TAILQ_INIT(&tactics->udpspeed.up_head);
    json_arr = cJSON_GetObjectItem(json, "udpUploadTactics");
    if (cJSON_IsArray(json_arr)) {
        pthread_mutex_init(&tactics->udpspeed.mutex, NULL);
        pthread_cond_init(&tactics->udpspeed.cond, NULL);
        tactics->udpspeed.up_count = cJSON_GetArraySize(json_arr);
        _parse_udpspeed_tactics(json_arr, &tactics->udpspeed.down_head, "up");
    }
    return 1;
}
static void _merge_data_to_reuslt(cJSON* result, cJSON* dial_result, char* name)
{
    cJSON *json_arr = NULL, *json_elem = NULL, *data_arr = NULL;
    int len;

    data_arr = cJSON_DetachItemFromObject(dial_result, name);

    if (!data_arr){
        cJSON_Delete(dial_result);
        return;
    }
        
    len = cJSON_GetArraySize(data_arr);
    if(len == 0){
        cJSON_Delete(data_arr);
        cJSON_Delete(dial_result);
        return;
    }

    json_arr = cJSON_GetObjectItem(result, name);
    if (json_arr) {
        while (len--) {
            json_elem = cJSON_DetachItemFromArray(data_arr, 0);
            cJSON_AddItemToArray(json_arr, json_elem);
        }
        //删除空数组
        cJSON_Delete(data_arr);
        cJSON_Delete(dial_result);
    } else {
        cJSON_AddItemToObject(result, name, data_arr);
        cJSON_Delete(dial_result);
    }

    log_trace("Append %s dial result json to dialtask root.", name);
}
static void _do_each_dial_tactics(struct Dial_Task_s* dialtask, Dial_Tactics_t* tactics, thread_pool_t* tpool)
{
    int i = 0, ii = 0, ret = 0;
    cJSON *result = NULL, *json_ptr = NULL;
    pthread_t thids[16] = { 0 };

    tactics->ping.taskId = dialtask->id;
    tactics->ping.version = dialtask->version;
    tactics->ping.kafka_url = dialtask->kafka_url;
    if (tactics->ping.count) {
        log_trace("pthread create ping task");
        ret = pthread_create(&thids[i++], NULL, do_ping_tactics, (void*)&tactics->ping);
        if(ret != 0){
            log_error("pthread create ping failed.%d,%s", ret, strerror(ret));
        }
        log_trace("pthread create ping task ok.");
    }

    tactics->tcp.taskId = dialtask->id;
    tactics->tcp.version = dialtask->version;
    tactics->tcp.kafka_url = dialtask->kafka_url;
    if (tactics->tcp.count) {
        log_trace("pthread create tcp task");
        ret = pthread_create(&thids[i++], NULL, do_tcp_tactics, (void*)&tactics->tcp);
        if(ret != 0){
            log_error("pthread create tcp failed.%d,%s", ret, strerror(ret));
        }
        log_trace("pthread create tcp task ok.");
    }

    tactics->dns.taskId = dialtask->id;
    tactics->dns.version = dialtask->version;
    tactics->dns.kafka_url = dialtask->kafka_url;
    if (tactics->dns.count) {
        log_trace("pthread create dns task");
        ret = pthread_create(&thids[i++], NULL, do_dns_tactics, (void*)&tactics->dns);
        if(ret != 0){
            log_error("pthread create dns failed.%d,%s", ret, strerror(ret));
        }
        log_trace("pthread create dns task ok.");
    }

    tactics->http.taskId = dialtask->id;
    tactics->http.version = dialtask->version;
    tactics->http.kafka_url = dialtask->kafka_url;
    if (tactics->http.count) {
        log_trace("pthread create http task");
        ret = pthread_create(&thids[i++], NULL, do_http_tactics, (void*)&tactics->http);
        if(ret != 0){
            log_error("pthread create http failed.%d,%s", ret, strerror(ret));
        }
        log_trace("pthread create http task ok.");
    }

    tactics->traceroute.taskId = dialtask->id;
    tactics->traceroute.version = dialtask->version;
    tactics->traceroute.kafka_url = dialtask->kafka_url;
    if (tactics->traceroute.count) {
        log_trace("pthread create traceroute task");
        ret = pthread_create(&thids[i++], NULL, do_traceroute_tactics, (void*)&tactics->traceroute);
        if(ret != 0){
            log_error("pthread create traceroute failed.%d,%s", ret, strerror(ret));
        }
        log_trace("pthread create traceroute task ok.");
    }

    for (ii = 0; ii < i; ii++) {
        if(thids[ii] == 0){
            continue;
        }

        pthread_join(thids[ii], (void*)&result);

        if(result == NULL){
            continue;
        }
        json_ptr = cJSON_GetObjectItem(result, "dnsList");
        if (json_ptr) {
            //_merge_data_to_reuslt 可能会释放 result 指针
            _merge_data_to_reuslt(dialtask->root, result, "dnsList");
            continue;
        }

        json_ptr = cJSON_GetObjectItem(result, "tcpList");
        if (json_ptr) {
            _merge_data_to_reuslt(dialtask->root, result, "tcpList");
            continue;
        }

        json_ptr = cJSON_GetObjectItem(result, "pingList");
        if (json_ptr) {
            _merge_data_to_reuslt(dialtask->root, result, "pingList");
            continue;
        }

        json_ptr = cJSON_GetObjectItem(result, "httpList");
        if (json_ptr) {
            _merge_data_to_reuslt(dialtask->root, result, "httpList");
            continue;
        }

        json_ptr = cJSON_GetObjectItem(result, "tracerouteList");
        if (json_ptr) {
            _merge_data_to_reuslt(dialtask->root, result, "tracerouteList");
            continue;
        }
    }
}
static void _merge_udpspeed_result_to_httpspeed(cJSON *root, cJSON *udpSpeedList){
    cJSON *httpSpeedList = NULL, *httpspeed = NULL, *udpspeed = NULL;
    cJSON *cjson_ptr = NULL;
    int httpspeed_len = 0;

    httpSpeedList = cJSON_GetObjectItem(root, "httpSpeedList");
    if(!httpSpeedList){
        return;
    }

    httpspeed_len = cJSON_GetArraySize(httpSpeedList);
    if(httpspeed_len == 0){
        return;
    }

    httpspeed = cJSON_GetArrayItem(httpSpeedList, httpspeed_len - 1);
    if(!httpspeed){
        return;
    }

    if(cJSON_GetObjectItem(httpspeed, "downMinSpeed") != NULL){
        return;
    }

    udpspeed = cJSON_GetArrayItem(udpSpeedList, 0);
    if(!udpspeed){
        cjson_ptr = cJSON_DetachItemFromArray(httpSpeedList, httpspeed_len - 1);
        cJSON_Delete(cjson_ptr);
        return;
    }

    cjson_ptr = cJSON_GetObjectItem(udpspeed, "downMinSpeed");
    cJSON_AddStringToObject(httpspeed, "downMinSpeed", cjson_ptr->valuestring);
    cjson_ptr = cJSON_GetObjectItem(udpspeed, "downMaxSpeed");
    cJSON_AddStringToObject(httpspeed, "downMaxSpeed", cjson_ptr->valuestring);
    cjson_ptr = cJSON_GetObjectItem(udpspeed, "downAvgSpeed");
    cJSON_AddStringToObject(httpspeed, "downAvgSpeed", cjson_ptr->valuestring);
    cjson_ptr = cJSON_GetObjectItem(udpspeed, "downActualDuration");
    cJSON_AddNumberToObject(httpspeed, "downActualDuration", cjson_ptr->valueint);
    cjson_ptr = cJSON_GetObjectItem(udpspeed, "downActualSize");
    cJSON_AddNumberToObject(httpspeed, "downActualSize", cjson_ptr->valueint);
}
static void _do_speed_tactics(struct Dial_Task_s* dialtask, httpspeed_tactics_t* httpspeed, udpspeed_tactics_t* udpspeed, httpvideodown_tactics_t* httpvideodown)
{
    cJSON *result = NULL, *json_ptr = NULL;

    httpspeed->taskId = dialtask->id;
    httpspeed->version = dialtask->version;
    httpspeed->kafka_url = dialtask->kafka_url;
    if (httpspeed->down_count || httpspeed->up_count) {
        result = do_httpspeed_tactics(httpspeed, dialtask->nodeUrl, dialtask->probeId);
        if(result != NULL){
            json_ptr = cJSON_GetObjectItem(result, "httpSpeedList");
            if (json_ptr){
                _merge_data_to_reuslt(dialtask->root, result, "httpSpeedList");
            }
        }
    }

    udpspeed->taskId = dialtask->id;
    udpspeed->version = dialtask->version;
    udpspeed->kafka_url = dialtask->kafka_url;
    if (udpspeed->down_count || udpspeed->up_count) {
        result = do_udpspeed_tactics(udpspeed, dialtask->nodeUrl, dialtask->probeId);
        if(result != NULL){
            json_ptr = cJSON_GetObjectItem(result, "udpSpeedList");
            if (json_ptr){
                //_merge_data_to_reuslt(dialtask->root, result, "udpSpeedList");
                _merge_udpspeed_result_to_httpspeed(dialtask->root, json_ptr);
                cJSON_Delete(result);
            }
        }
    }

    httpvideodown->taskId = dialtask->id;
    httpvideodown->version = dialtask->version;
    httpvideodown->kafka_url = dialtask->kafka_url;
    if (httpvideodown->count) {
        result = do_httpvideodown_tactics(httpvideodown, dialtask->nodeUrl, dialtask->probeId);
        if(result != NULL){
            json_ptr = cJSON_GetObjectItem(result, "httpVideoSpeedList");
            if (json_ptr){
                _merge_data_to_reuslt(dialtask->root, result, "httpVideoSpeedList");
            }
        }
    }
}
static void _do_each_tactics(struct Dial_Task_s* dialtask, Dial_Tactics_t* tactics)
{
    _do_each_dial_tactics(dialtask, tactics, NULL);

    _do_speed_tactics(dialtask, &tactics->httpspeed, &tactics->udpspeed, &tactics->httpvideodown);
}
static void _gen_sequnce_id(char* sequnce_str)
{


    char tmp_buf[32];
    int n = 0;
    struct tm *p, tm_p;
    time_t timep;

    time(&timep);
    localtime_r(&timep, &tm_p);

    p = &tm_p;
    if (currentMDay != p->tm_mday) {
        sequenceId = 0;
        currentMDay = p->tm_mday;
    }
    sequenceId += 1;

    sprintf(tmp_buf, "%d%02d%02d%02d%02d%02d",
        1900 + p->tm_year, 1 + p->tm_mon, p->tm_mday, p->tm_hour, p->tm_min, p->tm_sec);

    n += sprintf(sequnce_str + n, "%s", commReqParam.probeId);
    n += sprintf(sequnce_str + n, "%s", tmp_buf);
    n += sprintf(sequnce_str + n, "%05u", sequenceId);
}
static void _do_report(struct Dial_Task_s* dialtask)
{
    char timestap[32], sequnce_str[64];
    char* post_data;

    cJSON_AddStringToObject(dialtask->root, "dialTaskId", dialtask->id);
    cJSON_AddStringToObject(dialtask->root, "version", dialtask->version);
    cJSON_AddStringToObject(dialtask->root, "isdefault", dialtask->isdefault);
    cJSON_AddStringToObject(dialtask->root, "mac", devInfo.mac);
    cJSON_AddStringToObject(dialtask->root, "clientFactory", devInfo.factory);
    cJSON_AddStringToObject(dialtask->root, "clientModel", devInfo.model);
    cJSON_AddStringToObject(dialtask->root, "osType", devInfo.osType);
    cJSON_AddStringToObject(dialtask->root, "osVersion", devInfo.osVersion);
    cJSON_AddStringToObject(dialtask->root, "cityName", commReqParam.cityName);
    cJSON_AddStringToObject(dialtask->root, "cityAlias", commReqParam.cityAlias);
    cJSON_AddStringToObject(dialtask->root, "cityId", commReqParam.cityId);
    cJSON_AddStringToObject(dialtask->root, "province", commReqParam.province);
    cJSON_AddStringToObject(dialtask->root, "probeId", commReqParam.probeId);
    cJSON_AddStringToObject(dialtask->root, "probeVersion", commReqParam.version);
    cJSON_AddStringToObject(dialtask->root, "bdaccount", commReqParam.bdaccount);
    cJSON_AddStringToObject(dialtask->root, "realAccount", commReqParam.dial_bdaccount);
    cJSON_AddStringToObject(dialtask->root, "bdcode", commReqParam.bdcode);
    get_timestring(timestap);
    cJSON_AddStringToObject(dialtask->root, "reportTime", timestap);

    _gen_sequnce_id(sequnce_str);
    cJSON_AddStringToObject(dialtask->root, "sequenceId", sequnce_str);

    log_trace("dialtask->root:%p", dialtask->root);
    post_data = cJSON_Print(dialtask->root);
    
    if(Debug_mode()){
        log_trace("Report data:%s", post_data);
    }

    data_report(dialtask->kafka_url, dialtask->kafka_topic, post_data, strlen(post_data));

    free(post_data);
}

static void* _start_task(void* arg)
{
    struct Dial_Task_s* dialtask;
    Dial_Tactics_t* tactics;
    time_t now, done_tm = 0, last_report_tm = 0, delay_sec = 0, delay_tm = 0;
    int flag = 0;

    dialtask = (struct Dial_Task_s*)arg;
    dialtask->root = NULL;
    tactics = &dialtask->tactics;

    if(Debug_mode()){
        delay_sec = 0;
        delay_tm = 0;
    }else{
        delay_sec = 1 + (int)(get_random()/(RAND_MAX + 1.0) * dialtask->collectTime);
        delay_tm = time(NULL) + delay_sec;
    }
    
    
    log_info("Dial Task(id:%s, version:%s) Pid:%u, Tid:%u, delay:%d", dialtask->id, dialtask->version, getpid(), pthread_self(), delay_sec);

    dialtask->status = DIAL_TASK_EXIT_NONE;

    while (1) {

        sleep(1);
        now = time(NULL);

        switch (dialtask->flag) {
        case DIAL_TASK_FLAG_CANCEL:
            log_info("Dial Task(id:%s, version:%s) canceled.", dialtask->id, dialtask->version);
            dialtask->status = DIAL_TASK_EXIT_CANCELED;
            goto out;
        case DIAL_TASK_FLAG_STOP:
            continue;
        case DIAL_TASK_FLAG_RUN:
        default:
            break;
        }

        if(now < delay_tm){
            continue;
        }

        if (now > dialtask->endTime) { //over time
            log_info("Dial Task(id:%s, version:%s) is over time，now:%u, endTime:%u ",
                dialtask->id, dialtask->version, now, dialtask->endTime);
                dialtask->status = DIAL_TASK_EXIT_OVERTIME;
            goto out;
        }

        if (now < dialtask->startTime) {
            if (flag == 0) {
                log_warn("Dial Task(id:%s, version:%s) wait to startTime, now:%ld, startTime:%ld",dialtask->id, dialtask->version, now, dialtask->startTime);
                flag = 1;
            }
            continue;
        }

        if (now >= dialtask->startTime && now <= dialtask->endTime) {

            //是否到拨测时间
            if (now - done_tm > dialtask->collectTime) {
                if (dialtask->root == NULL) {
                    dialtask->root = cJSON_CreateObject(); //拨测数据会添加到这里
                }
                log_info("Task(id:%s, version:%s, done:%d)do each tactics", dialtask->id, dialtask->version, dialtask->done_count);
                done_tm = time(NULL);
                _do_each_tactics(dialtask, tactics);
                dialtask->done_count++;
            }
            //是否到上报时间
            if (now - last_report_tm > dialtask->cycleTime) {
                last_report_tm = time(NULL);
                if (dialtask->root != NULL) {
                    log_info("Task(id:%s, version:%s, report:%d)do report.", dialtask->id, dialtask->version, dialtask->done_report);
                    _do_report(dialtask);
                    cJSON_Delete(dialtask->root); //上报完后清除数据
                    dialtask->root = NULL;
                    dialtask->done_report++;
                }else{
                    // log_trace("Task(id:%s, version:%s) now:%d, last_report_tm:%d, cycleTime:%d.", dialtask->id, dialtask->version, now, last_report_tm, dialtask->cycleTime);
                    log_info("Task(id:%s, version:%s) Nothing to report.", dialtask->id, dialtask->version);
                }
            }
        }
    }

out:
    return (void*)dialtask->status;
}
static void _dial_task_notify_all_canceled(uint32_t cmd)
{
    struct sockaddr_in addr;
    int fd;

    sockaddr_set(&addr, AF_INET, htons(HEART_LISTNE_PORT), "127.0.0.1");
    fd = socket_stream_new();
    if (!fd) {
        log_syserr("dial task notify all canceled create socket error.");
        return;
    }
    if (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
        log_syserr("dial task notify all canceled connect error.");
        goto errout;
    }
    write(fd, &cmd, sizeof(cmd));
    log_trace("notify All task canceled.");

errout:
    close(fd);
    return;
}
void* dial_task_run(void* arg)
{

    int i = 0, ii = 0;
    int ret_code = -1, cancel_flg = 0;
    pthread_t thids[64] = { 0 };
    struct Dial_Task* taskall = (struct Dial_Task*)arg;
    struct Dial_Task_Head* head = &taskall->head;
    struct Dial_Task_s* dialtask;
    TAILQ_FOREACH(dialtask, head, entry)
    {
        pthread_create(&thids[i++], NULL, _start_task, (void*)dialtask);
    }
    for (ii = 0; ii < i; ii++) {
        pthread_join(thids[ii], (void*)&ret_code);
        log_trace("ret_code:%d", ret_code);
        if(ret_code == DIAL_TASK_EXIT_CANCELED){
            cancel_flg = 1;
        }
    }
    //如果存在，并且dialtask 的flag 为DIAL_TASK_FLAG_CANCEL，发送命令21
    log_debug("cancel_flg:%d", cancel_flg);
    if (cancel_flg) {
        _dial_task_notify_all_canceled(21); //通过发送21命令告知所有线程退出
    }
    return NULL;
}
// void dial_task_replace_by_id(struct Dial_Task_s* task, struct Dial_Task_Head* head)
// {
//     struct Dial_Task_s* ptr;

//     TAILQ_FOREACH(ptr, head, entry)
//     {
//         if (!strcmp(task->id, ptr->id)) {
//             TAILQ_REMOVE(head, ptr, entry);
//             dial_task_free(ptr);
//             TAILQ_INSERT_TAIL(head, task, entry);
//             break;
//         }
//     }
// }
void dial_task_free(struct Dial_Task_s* task)
{
    Dial_Tactics_t* tactics = &task->tactics;

    if(task->kafka_url){
        free(task->kafka_url);
    }

    if (task->root != NULL) {
        cJSON_Delete(task->root);
        task->root = NULL;
    }

    struct pingTactics* ping;
    TAILQ_FOREACH(ping, &tactics->ping.head, entry)
    {
        TAILQ_REMOVE(&tactics->ping.head, ping, entry);
        free(ping);
    }
    struct tcpTactics* tcp;
    TAILQ_FOREACH(tcp, &tactics->tcp.head, entry)
    {
        TAILQ_REMOVE(&tactics->tcp.head, tcp, entry);
        free(tcp);
    }
    struct dnsTactics* dns;
    TAILQ_FOREACH(dns, &tactics->dns.head, entry)
    {
        TAILQ_REMOVE(&tactics->dns.head, dns, entry);
        free(dns);
    }
    struct tracerouteTactics* traceroute;
    TAILQ_FOREACH(traceroute, &tactics->traceroute.head, entry)
    {
        TAILQ_REMOVE(&tactics->traceroute.head, traceroute, entry);
        free(traceroute);
    }
    struct httpTactics* http;
    TAILQ_FOREACH(http, &tactics->http.head, entry)
    {
        TAILQ_REMOVE(&tactics->http.head, http, entry);
        free(http);
    }
    struct httpspeedTactics* httpspeed;
    TAILQ_FOREACH(httpspeed, &tactics->httpspeed.down_head, entry)
    {
        TAILQ_REMOVE(&tactics->httpspeed.down_head, httpspeed, entry);
        free(httpspeed);
    }
    httpspeed = NULL;
    TAILQ_FOREACH(httpspeed, &tactics->httpspeed.up_head, entry)
    {
        TAILQ_REMOVE(&tactics->httpspeed.up_head, httpspeed, entry);
        free(httpspeed);
    }
    struct udpspeedTactics* udpspeed;
    TAILQ_FOREACH(udpspeed, &tactics->udpspeed.down_head, entry)
    {
        TAILQ_REMOVE(&tactics->udpspeed.down_head, udpspeed, entry);
        free(udpspeed);
    }
    udpspeed = NULL;
    TAILQ_FOREACH(udpspeed, &tactics->udpspeed.up_head, entry)
    {
        TAILQ_REMOVE(&tactics->udpspeed.up_head, udpspeed, entry);
        free(httpspeed);
    }
    free(task);
}
int dial_task_get_running_count(struct Dial_Task* taskall)
{
    struct Dial_Task_Head* head = &taskall->head;
    struct Dial_Task_s* ptr;
    int count = 0;

    TAILQ_FOREACH(ptr, head, entry)
    {
        if(ptr->status == DIAL_TASK_EXIT_NONE){
            log_notice("Task(id:%s, version:%s)running.", ptr->id, ptr->version);
            count++;
        }
    }
    log_trace("dial task running count:%d", count);
    return count;
}

void dial_task_free_all(struct Dial_Task* taskall)
{
    struct Dial_Task_Head* head = &taskall->head;
    struct Dial_Task_s* ptr;

    TAILQ_FOREACH(ptr, head, entry)
    {
        TAILQ_REMOVE(head, ptr, entry);
        dial_task_free(ptr);
    }
}
